package stc.skymobi.netty.client;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;

import org.apache.commons.collections.Closure;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import stc.skymobi.netty.handler.codec.memcached.*;
import stc.skymobi.netty.handler.codec.memcached.bean.request.RetrievalRequest;
import stc.skymobi.netty.handler.codec.memcached.bean.request.StorageRequest;
import stc.skymobi.netty.util.IntegerUUID;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.util.Date;
import java.util.concurrent.*;

import static stc.skymobi.netty.handler.codec.memcached.bean.core.BinaryCommonHeader.*;

public class MemCachedClient {
    public static Logger logger = LoggerFactory.getLogger(MemCachedClient.class);

    /**
     * values for cache flags
     */
    public static final int MARKER_BYTE = 1;
    public static final int MARKER_BOOLEAN = 8192;
    public static final int MARKER_INTEGER = 4;
    public static final int MARKER_LONG = 16384;
    public static final int MARKER_CHARACTER = 16;
    public static final int MARKER_STRING = 32;
    public static final int MARKER_STRINGBUFFER = 64;
    public static final int MARKER_FLOAT = 128;
    public static final int MARKER_SHORT = 256;
    public static final int MARKER_DOUBLE = 512;
    public static final int MARKER_DATE = 1024;
    public static final int MARKER_STRINGBUILDER = 2048;
    public static final int MARKER_BYTEARR = 4096;
    public static final int MARKER_OTHERS = 0x00;

    private String defaultEncoding = "utf-8";
    private TransCoder transCoder = new ObjectTransCoder();

    // flags
    private boolean sanitizeKeys;
    private boolean primitiveAsString;

    private long opTimeOut;
    private int maxHoldSize;
    private int maxSenderThreads;

    // optional error handler
    protected ErrorHandler errorHandler;
    private RequestSender sender;

    private BlockingQueue requestQueue;
    private Cache<Integer, Closure> uuidCache;

    public MemCachedClient() {
        this(10, 10000, 60000);
    }

    public MemCachedClient(int maxSenderThreads, int maxHoldSize, long opTimeOut) {
        this.opTimeOut = opTimeOut;
        this.maxSenderThreads = maxSenderThreads;
        this.maxHoldSize = maxHoldSize;

        uuidCache = CacheBuilder.newBuilder().expireAfterWrite(opTimeOut, TimeUnit.MILLISECONDS).build();
        requestQueue = new ArrayBlockingQueue(maxHoldSize);
    }

    public void init() {
        ExecutorService worker = new ThreadPoolExecutor(0, maxSenderThreads, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
        for (int i = 0; i < maxSenderThreads; i++) {
            worker.submit(sender);
        }
    }

    public void set(String key, Object value, Closure callback) {
        set(OPCODE_SET, key, value, null, null, 0L, primitiveAsString, callback);
    }

    public void set(String key, Object value, Integer hashCode, Closure callback) {
        set(OPCODE_SET, key, value, null, hashCode, 0L, primitiveAsString, callback);
    }

    public void set(String key, Object value, Date expiry, Closure callback) {
        set(OPCODE_SET, key, value, expiry, null, 0L, primitiveAsString, callback);
    }

    public void set(String key, Object value, Date expiry, Integer hashCode, Closure callback) {
        set(OPCODE_SET, key, value, expiry, hashCode, 0L, primitiveAsString, callback);
    }

    public void add(String key, Object value, Closure callback) {
        set(OPCODE_ADD, key, value, null, null, 0L, primitiveAsString, callback);
    }

    public void add(String key, Object value, Integer hashCode, Closure callback) {
        set(OPCODE_ADD, key, value, null, hashCode, 0L, primitiveAsString, callback);
    }

    public void add(String key, Object value, Date expiry, Closure callback) {
        set(OPCODE_ADD, key, value, expiry, null, 0L, primitiveAsString, callback);
    }

    public void add(String key, Object value, Date expiry, Integer hashCode, Closure callback) {
        set(OPCODE_ADD, key, value, expiry, hashCode, 0L, primitiveAsString, callback);
    }

    public void get(String key, Closure callback) {
        get(key, null, callback);
    }

    public void get(String key, Integer hashCode, Closure callback) {
        get(OPCODE_GET, key, hashCode, false, callback);
    }

    public void get(String key, Integer hashCode, boolean asString, Closure callback) {
        get(OPCODE_GET, key, hashCode, asString, callback);
    }

    /**
     * Set, Add, Replace data to cache.
     * <p/>
     * If data does not already exist for this key on the server, or if the key
     * is being<br/>
     * deleted, the specified value will not be stored.<br/>
     * The server will automatically delete the value when the expiration time
     * has been reached.<br/>
     * <br/>
     * If compression is enabled, and the data is longer than the compression
     * threshold<br/>
     * the data will be stored in compressed form.<br/>
     * <br/>
     * As of the current release, all objects stored will use java
     * serialization.
     *
     * @param key      key to store cache under
     * @param value    object to cache
     * @param expiry   expiration
     * @param hashCode if not null, then the int hashcode to use
     * @return true/false indicating success
     */
    private void set(byte opCode, String key, Object value, Date expiry, Integer hashCode, long casUnique, boolean asString, Closure callback) {
        StorageRequest request = new StorageRequest();
        request.setHashCode(hashCode);
        request.setCallback(callback);

        if (key == null) {
            logger.error("key is null or cmd is null/empty for set()");
            callback.execute(null);
            return;
        }

        try {
            key = sanitizeKey(key);
        } catch (UnsupportedEncodingException e) {

            // if we have an errorHandler, use its hook
            if (errorHandler != null)
                errorHandler.handleErrorOnSet(this, e, key);

            logger.error("failed to sanitize your key!", e);
            callback.execute(null);
            return;
        }

        if (value == null) {
            logger.error("trying to store a null value to cache");
            callback.execute(null);
            return;
        }

        if (expiry == null)
            expiry = new Date(0);

        try {
            // store flags
            int flags = NativeHandler.getMarkerFlag(value);
            byte[] buf = key.getBytes();

            //sock.writeBuf.put(MAGIC_REQ);
            request.setOpCode(opCode);
            request.setKeyLength((short) buf.length);
            request.setExtrasLength((byte) 0x08);
            request.setDataType((byte) 0);
            request.setVbucketId((short) 0);
            request.setOpaque(IntegerUUID.nextUUID());
            request.setCas(casUnique);
            request.setFlags(flags);
            request.setExpiration(new Long(expiry.getTime() / 1000).intValue());
            request.setKey(key);
            ByteArrayOutputStream output = new ByteArrayOutputStream();
            int valLen = 0;
            if (flags != MARKER_OTHERS) {
                byte[] b;
                if (asString) {
                    b = value.toString().getBytes(defaultEncoding);
                } else {
                    /*
                          * Using NativeHandler to serialize the value
                          */
                    b = NativeHandler.encode(value);
                }
                request.setValue(b);
                valLen = b.length;
                valLen = b.length;

            } else {
                // always serialize for non-primitive types
                valLen = transCoder.encode(output, value);
            }
            // write serialized object
            int bodyLen = 0x08 + buf.length + valLen;
            // put real object bytes size
            request.setTotalBodyLength(bodyLen);
            requestQueue.put(request);
        } catch (IOException e) {

            // if we have an errorHandler, use its hookE
            if (errorHandler != null)
                errorHandler.handleErrorOnSet(this, e, key);

            // exception thrown
            logger.error("++++ exception thrown while writing bytes to server on set");
            logger.error(e.getMessage(), e);
        } catch (InterruptedException e) {
            logger.error("", e);
        }
    }

    private String sanitizeKey(String key) throws UnsupportedEncodingException {
        return (sanitizeKeys) ? URLEncoder.encode(key, "UTF-8") : key;
    }

    /**
     * Retrieve a key from the server, using a specific hash.
     * <p/>
     * If the data was compressed or serialized when compressed, it will
     * automatically<br/>
     * be decompressed or serialized, as appropriate. (Inclusive or)<br/>
     * <br/>
     * Non-serialized data will be returned as a string, so explicit conversion
     * to<br/>
     * numeric types will be necessary, if desired<br/>
     *
     * @param key      key where data is stored
     * @param hashCode if not null, then the int hashcode to use
     * @param asString if true, then return string val
     * @return the object that was previously stored, or null if it was not
     *         previously stored
     */
    private void get(byte opCode, String key, Integer hashCode, boolean asString, Closure callback) {
        if (key == null) {
            logger.error("key is null for get()");
            callback.execute(null);
            return;
        }

        try {
            key = sanitizeKey(key);
        } catch (UnsupportedEncodingException e) {
            // if we have an errorHandler, use its hook
            if (errorHandler != null)
                errorHandler.handleErrorOnGet(this, e, key);

            logger.error("failed to sanitize your key!", e);
            callback.execute(null);
            return;
        }

        try {
            RetrievalRequest request = new RetrievalRequest();
            request.setHashCode(hashCode);
            request.setCallback(callback);
            request.setAsString(asString);
            byte[] buf = key.getBytes();
            request.setOpCode(opCode);
            request.setKeyLength((short) buf.length);
            request.setExtrasLength((byte) 0);
            request.setDataType((byte) 0);
            request.setVbucketId((short) 0);
            request.setOpaque(IntegerUUID.nextUUID());
            request.setCas(0);
            request.setKey(key);
            request.setTotalBodyLength(buf.length);
            // write buffer to server
            requestQueue.put(request);

        } catch (InterruptedException e) {
            logger.error("", e);
        }
    }

    public void setDefaultEncoding(String defaultEncoding) {
        this.defaultEncoding = defaultEncoding;
    }

    public void setSanitizeKeys(boolean sanitizeKeys) {
        this.sanitizeKeys = sanitizeKeys;
    }

    public void setPrimitiveAsString(boolean primitiveAsString) {
        this.primitiveAsString = primitiveAsString;
    }

    public void setErrorHandler(ErrorHandler errorHandler) {
        this.errorHandler = errorHandler;
    }

    public void setTransCoder(TransCoder transCoder) {
        this.transCoder = transCoder;
    }

    public BlockingQueue getRequestQueue() {
        return requestQueue;
    }

    public Cache<Integer, Closure> getUuidCache() {
        return uuidCache;
    }

    public void setOpTimeOut(long opTimeOut) {
        this.opTimeOut = opTimeOut;
    }

    public int getMaxSenderThreads() {
        return maxSenderThreads;
    }

    public void setMaxSenderThreads(int maxSenderThreads) {
        this.maxSenderThreads = maxSenderThreads;
    }

    public void setSender(RequestSender sender) {
        this.sender = sender;
    }
}
